#encoding:utf-8
#author:donghao

import time
import redis
import json
import logging
from canal.client import Client
from canal.protocol import EntryProtocol_pb2

logging.basicConfig(
    filename="info.log",
    format='%(asctime)s-%(name)s-%(levelname)s-%(module)s:%(message)s',
    level=10,
    filemode="w",
    datefmt='%m/%d/%Y %I:%M:%S'
)

client = Client()
client.connect(host='192.168.0.180', port=11111)
client.check_valid(username=b'', password=b'')
client.subscribe(client_id=b'1001', destination=b'example', filter=b'.*\\..*')
pool = redis.ConnectionPool(host='192.168.0.180', port=6379, password='')
cache = redis.Redis(connection_pool=pool)
while True:
    message = client.get(100)
    entries = message['entries']
    for entry in entries:
        entry_type = entry.entryType
        if entry_type in [EntryProtocol_pb2.EntryType.TRANSACTIONBEGIN, EntryProtocol_pb2.EntryType.TRANSACTIONEND]:
            continue
        row_change = EntryProtocol_pb2.RowChange()
        row_change.MergeFromString(entry.storeValue)
        event_type = row_change.eventType
        header = entry.header
        database = header.schemaName
        table = header.tableName
        event_type = header.eventType
        for row in row_change.rowDatas:
            format_data = dict()
            if event_type == EntryProtocol_pb2.EventType.DELETE:
                for column in row.beforeColumns:
                    format_data[column.name] = column.value
            elif event_type == EntryProtocol_pb2.EventType.INSERT:
                for column in row.afterColumns:
                    format_data[column.name] = column.value
            else:
                format_data['before'] = format_data['after'] = dict()
                for column in row.beforeColumns:
                    format_data['before'][column.name] = column.value
                for column in row.afterColumns:
                    format_data['after'][column.name] = column.value
            data = dict(
                db=database,
                table=table,
                event_type=event_type,
                data=format_data,
            )
            print(data)
            if event_type == EntryProtocol_pb2.EventType.UPDATE:
                logging.info("修改数据--table name:{}---data:{}".format(table,data))
                # 更新前
                cache_key = "{}:id_{}".format(table,format_data['before']['id'])
                cached_article = cache.get(cache_key)
                if cached_article:
                    print('更新前 ', json.loads(cached_article))
                # 更新缓存
                cache.set(cache_key, json.dumps(format_data['after']))
                print('更新后 ', json.loads(cache.get(cache_key)))
            elif event_type == EntryProtocol_pb2.EventType.INSERT:
                logging.info("添加数据--table name:{}---data:{}".format(table, data))
                cache_key = "{}:id_{}".format(table, format_data['id'])
                cached_article = cache.get(cache_key)
                if cached_article:
                    print('更新前 ', json.loads(cached_article))
                # 更新缓存
                cache.set(cache_key, json.dumps(format_data))
                print('更新后 ', json.loads(cache.get(cache_key)))
            elif event_type == EntryProtocol_pb2.EventType.DELETE:
                logging.info("删除数据--table name:{}---data:{}".format(table, data))
                cache_key = "{}:id_{}".format(table, format_data['id'])
                cached_article = cache.get(cache_key)
                if cached_article:
                    cache.delete(cache_key)
    time.sleep(1)
client.disconnect()
